package com.lagou.hbase;

import jdk.nashorn.internal.ir.CallNode;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import java.io.IOException;
import java.util.List;

/**
 * @Author Cao Dun
 * @Date 2021/3/24 22:26
 * @Version 1.0
 * @Description 删除好友时，互相删除
 */
public class DeleteFriendProcessor extends BaseRegionObserver {


    private HTableWrapper  friends;

    private static final Logger logger = Logger.getLogger(DeleteFriendProcessor.class);

    /**
     *  查询某个列族中value对应的列名
     * @param rowKey
     * @param colFamily
     * @param targetValue
     * @return 返回列名
     * @throws IOException
     */
    private String getValueCol(String rowKey,String colFamily,String targetValue) throws IOException {
        String targetColumn=null;

        Get get = new Get(rowKey.getBytes());
        get.addFamily(colFamily.getBytes());
        //按列值过滤
        get.setFilter(new ValueFilter(CompareFilter.CompareOp.EQUAL,new BinaryComparator(targetValue.getBytes())));

        Result result = friends.get(get);
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.println(cell);
            String cf=Bytes.toString(CellUtil.cloneFamily(cell));
            String column=Bytes.toString(CellUtil.cloneQualifier(cell));
            String value=Bytes.toString(CellUtil.cloneValue(cell));
            if (value.equals(targetValue)){
                targetColumn=column;
                continue;
            }
        }
        return targetColumn;
    }

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        //获取到hbase表对象
        friends = (HTableWrapper )e.getEnvironment().getTable(TableName.valueOf("friends"));
        //获取删除的行,就是删除了好友的人的id，被它删除的人需要删除它
        byte[] row = delete.getRow();
        //所有被删除的cell
        List<Cell> delCells = delete.getFamilyCellMap().get(Bytes.toBytes("info"));
        for (Cell delCell : delCells) {
            //获取到被删除的列名
            byte[] delQualify= CellUtil.cloneQualifier(delCell);
            logger.info("delQualify="+delQualify.toString());
            //获取被删除的列值
            byte[] delValue = CellUtil.cloneValue(delCell);
            System.out.println(String.valueOf(delValue));

            //找到row对应的列名(发起删除的rowkey，在他的好友的记录中对应的列是什么)
            //String info = "info";
            //String valueCol = getValueCol(delValue.toString(), info, row.toString());

            //删除
            //Delete delete1 = new Delete(delValue);
            //delete1.addColumn("info".getBytes(),valueCol.getBytes());
            //friends.delete(delete1);
        }
        friends.flushCommits();
        friends.close();


    }
}
